﻿using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading;
using Platform.Storage.Redis.Messages;

namespace Platform.Storage.Redis
{

	/// <summary>
	/// 
	/// </summary>
	public class Messaging : IDisposable
	{
		/// <summary>
		/// Define de quanto em quanto tempo será verificado se existe dados para serem processados
		/// </summary>
		private static readonly TimeSpan notificationTimeout = TimeSpan.FromMilliseconds( 100 );
		private RedisMessageChannel channel;
		private RedisConnection connection;
		private Thread listener;
		

		/// <summary>
		/// Evento lançado quando uma mensagems chegou no canal
		/// </summary>
		public event EventHandler<MessageReceivedEventArgs> Received;

		internal Messaging( RedisMessageChannel channel )
			: this( null, channel )
		{ }

		internal Messaging( RedisConnection connection )
			: this( connection, null )
		{ }

		private Messaging( RedisConnection connection, RedisMessageChannel channel )
		{

			if ( channel == null )
			{
				if ( connection == null )
					throw new ArgumentNullException( "channel" );
				this.channel = connection.Channel;
			}
			else
				this.channel = channel;
			this.connection = connection;
		}


		public int SubscribeIfMatches( params string[] channelPatterns )
		{
			return this.Subscribe( SubscriptionMatch.ByPattern, channelPatterns );
		}

		/// <summary>
		/// 
		/// </summary>
		/// <param name="channel"></param>
		/// <param name="channels"></param>
		/// <returns>Retorna o numero de canais que estao ativos</returns>
		public int Subscribe( params string[] channels )
		{
			return this.Subscribe( SubscriptionMatch.Exact, channels );
		}
		
		public int Subscribe( SubscriptionMatch match, string[] channels )
		{
			//int clients = this.channel.ExecuteInt32( "PSUBSCRIBE", channels );
			var subscription = this.channel.SendAndWaitReply( new SubscribeMessage( this.channel, match, channels ) );

			if ( this.listener == null )
			{
				Thread worker =
					new Thread(
						new ThreadStart(
							() =>
							{
								while ( this.channel.Connected )
								{
									ChannelNotificationMessage msg;

									if ( this.channel.WaitNotification( notificationTimeout, out msg ) )
										foreach ( var notification in msg.Reply )
											this.OnReceived( new MessageReceivedEventArgs( notification.Channel, notification.Message ) );
								}
							} ) );

				Interlocked.CompareExchange( ref this.listener, worker, null );
				// Se for a mesma thread que "setou" o listener, então somente ela pode inicializar o listener
				if ( Object.ReferenceEquals( this.listener, worker ) )	
					this.listener.Start();
			}

			return subscription.Reply.MonitoringCount;
		}

		/// <summary>
		/// 
		/// </summary>
		/// <param name="channels"></param>
		public void Unsubscribe( params string[] channels )
		{
			this.channel.ExecuteSingle( "PUNSUBSCRIBE", channels );
		}

		/// <summary>
		/// Envia a mensagem
		/// </summary>
		/// <param name="channel"></param>
		/// <param name="message"></param>
		/// <returns>Quantidade de clientes que receberam a mensagem</returns>
		public int Publish( string channelName, string message )
		{
			return this.channel.Send( new PublishMessage( this.channel, channelName, message ) ).Reply;
		}

		/// <summary>
		/// 
		/// </summary>
		/// <param name="args"></param>
		protected virtual void OnReceived( MessageReceivedEventArgs args )
		{
			if ( this.Received != null )
				this.Received( this, args );
		}

		#region IDisposable Members

		public void Dispose()
		{
			try
			{
				this.Unsubscribe();
			}
			finally
			{
				var worker = Interlocked.Exchange( ref this.listener, null );
				if ( worker != null )
					worker.Abort();

				if ( this.connection != null )
				{
					this.connection.Close();
					this.connection.Dispose();
				}
			}
		}

		#endregion
	}

}
